package com.hcy.schedulingapi.service.mqconsumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.hcy.common.dto.response.CallCarConfigDTO;
import com.hcy.common.dto.response.GaoDeGfidsResponseDTO;
import com.hcy.driver.DriverBaseInfoClient;
import com.hcy.driver.dto.request.DriverCarInfoRequestDTO;
import com.hcy.driver.dto.response.DriverCarInfoResponseDTO;
import com.hcy.mqclient.message.MessageServiceClient;
import com.hcy.mqmodel.dto.request.MessageListRequestDTO;
import com.hcy.orderclient.OrderForDriverClient;
import com.hcy.ordermodel.dto.request.DriverOrderRequestDTO;
import com.hcy.ordermodel.dto.response.AddOrderResponseDTO;
import com.hcy.ordermodel.dto.response.SimpleOrderResponseDTO;
import com.hcy.rabbitbasicmodel.common.MyWebSocketMessage;
import com.hcy.rabbitbasicmodel.enums.Clients;
import com.hcy.rabbitbasicmodel.enums.Code;
import com.hcy.rabbitbasicmodel.enums.M_Code;
import com.hcy.rabbitbasicmodel.model.ResponseData;
import com.hcy.rabbitbasicutils.utils.RedisUtils;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @Author huchenying
 * @Description
 * @Date 2022/3/15 15:55
 */
@Component
public class MQDispatcherOrderConsumer implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(MQDispatcherOrderConsumer.class);

    private static final ThreadLocal<DefaultMQPushConsumer> threadLocal = new ThreadLocal<DefaultMQPushConsumer>() {
        /**
         * ThreadLocal没有被当前线程赋值时或当前线程刚调用remove方法后调用get方法，返回此方法值
         */
        @Override
        protected DefaultMQPushConsumer initialValue() {
            return null;
        }
    };

    public static final ConcurrentHashMap<Long, GaoDeGfidsResponseDTO> mpInfo = new ConcurrentHashMap<>();

    @Autowired
    RedisUtils redisUtils;

    @Autowired
    MessageServiceClient messageServiceClient;

    @Autowired
    DriverBaseInfoClient driverBaseInfoClient;

    @Autowired
    OrderForDriverClient orderForDriverClient;

    @Override
    public void run() {
        // 如果为空则创建
        if (threadLocal.get() == null) {
            GaoDeGfidsResponseDTO gaoDeGfidsResponseDTO = mpInfo.get(Thread.currentThread().getId());
            logger.info("gaoDeGfids info -> {}", ToStringBuilder.reflectionToString(gaoDeGfidsResponseDTO));
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(gaoDeGfidsResponseDTO.getGfid().toString());
            //设置NameServer地址
            consumer.setNamesrvAddr(gaoDeGfidsResponseDTO.getAddress());
            //设置起始消费位置
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //设置消费超时时间
//            consumer.setConsumeTimeout(10 * 60 * 1000);
            //指定消费Topic和Tag

            try {
                consumer.subscribe(gaoDeGfidsResponseDTO.getGfid().toString(), "*");
            } catch (MQClientException e) {
                throw new RuntimeException("设置topic异常");
            }

            //指定消费模式 默认集群模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //设置消费启动线程
            consumer.setConsumeThreadMin(1);
            consumer.setConsumeThreadMax(1);
            //设置每批消费多少
            consumer.setConsumeMessageBatchMaxSize(10);
            threadLocal.set(consumer);
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {

                //消费一条消息
                logger.info("msgs size = [{}]", msgs.size());
                for (MessageExt msg : msgs) {
//                System.out.println(msg.getKeys());
                    try {
                        consumer(msg);
                    } catch (Exception e) {
                        logger.error("消费异常 e-> {}",ToStringBuilder.reflectionToString(e));
                    }
                    logger.info("消费成功！");
                }
                //返回消费状态
                //重新消费
//                ConsumeConcurrentlyStatus.RECONSUME_LATER
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            //开启消费
            try {
                consumer.start();
                logger.info("消费派单consumer启动成功！");
            } catch (MQClientException e) {
                e.printStackTrace();
                logger.error("消费派单consumer启动失败！");
            }
        }
    }

    /**
     * @Author huchenying
     * @Description 消费消息
     * @Date 2022/3/15 16:14
     **/
    public void consumer(MessageExt messageExt){
        logger.info("consumer start param -> {}", ToStringBuilder.reflectionToString(messageExt));
        //反序列化
        String keys = messageExt.getKeys();
        AddOrderResponseDTO serverOrder = null;
        serverOrder = JSON.parseObject(messageExt.getBody(), AddOrderResponseDTO.class);

        //在这做处理，可以直接人工干预
        if (serverOrder == null) {
            logger.error("serverOrder 空订单无法消费");
            throw new RuntimeException("serverOrder 空订单无法消费");
        }
        logger.info("消费订单消息json解析成功！serverOrder -> {}",ToStringBuilder.reflectionToString(serverOrder));
        //判断订单状态
        //首先判断订单是否已经取消
        if (redisUtils.isOrderNotKeepAlive(keys)) {
            logger.info("订单[{}]已经消费或取消，跳过消费", keys);
            return;
        }
        //这里可以从本地缓存中加载配置，决定叫车策略
        // 叫几次，每一次的策略 id,范围，据下一次间隔时间，呼叫完这一次后是否转到人工派单
        String call_car_config = redisUtils.get("CALL_CAR_CONFIG");
        List<CallCarConfigDTO> callCarConfigDTOS = JSONArray.parseArray(call_car_config, CallCarConfigDTO.class);
        for (CallCarConfigDTO callCarConfigDTO : callCarConfigDTOS) {
            logger.info("当前订单:{},第{}次呼叫", serverOrder.getOrderNumber(), callCarConfigDTO.getRound());
            String orderCache = redisUtils.getOrderCache(serverOrder.getOrderNumber());
            SimpleOrderResponseDTO simpleOrder = null;
            if(orderCache == null){ //缓存没有，需要查库
                DriverOrderRequestDTO requestDTO = new DriverOrderRequestDTO();
                requestDTO.setOrderNumber(serverOrder.getOrderNumber());
                ResponseData<SimpleOrderResponseDTO> responseData = orderForDriverClient.getSimpleOrder(requestDTO);
                if(responseData.getCode() != Code.Success){
                    logger.error("获取不到订单终止消费");
                    return;
                }
                simpleOrder = responseData.getData();
                //重新设置缓存
                redisUtils.setOrderCache(serverOrder.getOrderNumber(),JSON.toJSONString(simpleOrder),30, TimeUnit.MINUTES);
            }else{
                logger.info("订单{}缓存命中",serverOrder.getOrderNumber());
                simpleOrder = JSON.parseObject(orderCache,SimpleOrderResponseDTO.class);
            }
            //判断订单状态
            //首先判断订单是否已经取消
            if (redisUtils.isOrderNotKeepAlive(keys)) {
                logger.info("订单[{}]已经消费或取消，跳过消费", keys);
                return;
            }
            logger.info("订单{} Alive", keys);
            //查询当前订单信息
            String startLongitude = simpleOrder.getStartLongitude();
            String startLatitude = simpleOrder.getStartLatitude();
            //给乘客反馈
            if(callCarConfigDTO.getIsCallback().equals(1)){
                Message message = new Message();
                message.setTopic("WEBSOCKET");
                MyWebSocketMessage customerTips = new MyWebSocketMessage();
                customerTips.setCode(M_Code.CustomerCallBackInfo.getCode());
                customerTips.setClientId(Clients.CUSTOMER.getName());
                customerTips.setDesc(M_Code.CustomerCallBackInfo.getDesc());
                customerTips.setUsername(serverOrder.getPassengerPhone());
                customerTips.setData(callCarConfigDTO.getCallbackInfo());
                message.setBody(JSON.toJSONString(customerTips).getBytes());
                messageServiceClient.pushAsyncMessage(message);
            }
            //查询当前听单的司机,注意怎么放的，出了问题看一下前面怎么放的经纬度位置有没有搞反！
            //最好优化一下，筛选出未出车司机和已结单司机
            List<Integer> carList = redisUtils.searchGeo("CAR_GEO_KEY", Double.parseDouble(startLatitude), Double.parseDouble(startLongitude), callCarConfigDTO.getRadius());
            carList = redisUtils.selectDriver(carList,5);
            logger.info("查询上车点{}附近{}米可接单司机{}位", startLongitude +","+ startLatitude, callCarConfigDTO.getRadius(), carList.size());
            //如果有，则呼叫司机，发送websocekt消息，携带订单号，司机再去接口请求订单其他信息
            if (carList.size() != 0) {
//                // 根据配置截取list,可以在这做一些算法配置选取最优的司机
//                if (carList.size() > callCarConfigDTO.getPushMaxCar()) {
//                    carList = carList.subList(0, callCarConfigDTO.getPushMaxCar());
//                }
                ////通过carId找到司机信息
                DriverCarInfoRequestDTO driverCarInfoRequestDTO = new DriverCarInfoRequestDTO();
                driverCarInfoRequestDTO.setCarIds(carList);
                ResponseData<List<DriverCarInfoResponseDTO>> listResponseData = driverBaseInfoClient.carDriverInfoList(driverCarInfoRequestDTO);
                if (listResponseData.getCode() != Code.Success) {
                    logger.error("查询车辆司机信息失败，param - >{}", ToStringBuilder.reflectionToString(driverCarInfoRequestDTO));
                    return;
                }
                logger.info("发送websocketMessageList通知司机抢单");
                //发送websocet消息 Topic为 WEBSOCKET
                List<Message> messageList = new ArrayList<>();
                //可优化，批量发送
                for (DriverCarInfoResponseDTO carInfoResponseDTO : listResponseData.getData()) {
                    Message message = new Message();
                    message.setTopic("WEBSOCKET");
                    MyWebSocketMessage webSocketMessage = new MyWebSocketMessage();
                    webSocketMessage.setClientId(Clients.DRIVER.getName());
                    webSocketMessage.setDesc(M_Code.SystemCallDriverOrder.getDesc());
                    webSocketMessage.setUsername(carInfoResponseDTO.getDriverPhone());
                    webSocketMessage.setCode(M_Code.SystemCallDriverOrder.getCode());
                    // 设置消息体
                    SimpleOrderResponseDTO simpleOrderResponseDTO = new SimpleOrderResponseDTO();
                    BeanUtils.copyProperties(serverOrder, simpleOrderResponseDTO);
                    webSocketMessage.setData(simpleOrderResponseDTO);
                    message.setBody(JSON.toJSONString(webSocketMessage).getBytes());
                    messageList.add(message);
                }
                MessageListRequestDTO messageListRequestDTO = new MessageListRequestDTO();
                messageListRequestDTO.setMessageList(messageList);
                logger.info("发送消息给符合条件的司机 param -> {}",ToStringBuilder.reflectionToString(messageListRequestDTO));
                messageServiceClient.pushAsyncMessageList(messageListRequestDTO);
            }else {
                logger.info("未找到可接单司机");
            }
            // 间隔下一次呼叫多少秒
            try {
                Thread.sleep(callCarConfigDTO.getGapTime() * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 转入人工派单，或者持续循环
        logger.info("没有人接单。。");
    }
}
